package jccr;

import java.util.ArrayList;
import java.lang.InterruptedException;
import java.util.concurrent.BrokenBarrierException;
import java.util.concurrent.CyclicBarrier;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.locks.Condition;
import java.util.concurrent.locks.ReentrantLock;

public class ThreadPool implements Dispatcher, DispatcherQueueMonitor {
	
	public ThreadPool( String name, int count ) {
		
		if ( count <= 0 )
			throw new IllegalArgumentException();
				
		_name = name != null ? name : "ThreadPool";
		createWorkers( count );
		
		_running.set(true);
		startWorkers();
	}
			
	public void attach( DispatcherQueue queue ) {
		
		if ( queue == null )
			throw new IllegalArgumentException();
		
		assert( _queues != null );
		
		queue.monitor(this);
		
		_lock.lock();
		try { 
			_queues.add(queue); 
		}
		finally { 
			_lock.unlock(); 
		}			
	}
	
	public void dispose() throws InterruptedException {
		
		_running.set(false);	
		signalWorkers();
		
		/* wait for all threads to terminate; this reduces the risks of fooling
		 * the developer by running dispatched tasks on threads from the ThreadPool
		 * after dispose() has been called, however this will cause the dispose() method
		 * to block until all tasks have completed.
		 */
		for( Worker w : _workers ) 
			w.join();
	}
	
	/**
	 * Get the pool thread count.
	 * @return the count.
	 */
	public int getWorkerCount() { return _workers.length; }
	
	/**
	 * Gets the number of attached queues.
	 * @return the number of attached queues
	 */
	public int getQueueCount() { 
		
		assert( _queues != null );
		
		_lock.lock();
		try {
			return _queues.size();
		}
		finally {
			_lock.unlock();
		}				 
	}
	
	/**
	 * Gets the pool name.
	 * @return the name.
	 */
	public String getName() { return _name; }
	
	
	public boolean isIdle() {
		
		for( Worker w : _workers ) {
			
			if ( !w.isIdle() )
				return false;
		}
		
		return true;
	}
	
	public void taskEnqueued( DispatcherQueue queue ) {
	
		signalWorkers();		
	}	
	
	void createWorkers( int count ) {
		
		assert(_workers == null);
		assert(_name == null);
	
		_workers = new Worker[count];
		for( int i=0; i<count; i++ ) {
			
			String name = formatWorkerName(_name, i);			 				
			_workers[i] = new Worker( name );			
		}
	}
	
	DispatcherQueue getNextQueue() {				
		
		_lock.lock();
		try {
			
			int size = _queues.size();
			
			/* fast return to avoid division-by-zero if the queue size is 0
			 * The most likely reason for this return is a developer error.
			 */
			if ( size == 0 )
				return null;
						
			int index = _queueIndex;
			_queueIndex = ++_queueIndex % _queues.size();
			assert( index >= 0 && index < size );
			
			return _queues.get(index);
		}
		finally { _lock.unlock(); }
	}
	
	/**
	 * Formats the name of the worker based on the pool name and the worker count
	 * @param queueName the queue name
	 * @param count the worker count
	 * @return the formatted worker name
	 */
	static String formatWorkerName( String queueName, int count ) {
		return queueName + " " + count;
	}
	
	void signalWorkers() {
	
		for( Worker w : _workers ) {		
			w.signal();
		}
	}
	
	void startWorkers() {
		
		for( Worker worker : _workers ) {			
			worker.start();
		}
	}
	
	class Worker extends Thread {
		
		public Worker( String name ) {
			super(name);
		}
		
		public boolean isIdle() { return _idle; }
		
		@Override
		public void run() {
								
			while( _running.get() ) {
			
				/* attempt to get the next task, if this fails spin-wait 
				 * for a short while and then suspend the current thread
				 */
				DispatcherQueue queue = nextQueue();
				Task next = null;
				
				if ( queue != null && ( next = queue.dequeue() ) != null ) {
					execute( queue, next );
				} 
				else {
					
					/* if we hit the max spin count block the thread */
					if ( ++_spinCounter >= MAX_SPIN_COUNT ) {
						waitForSignal();
					}					
				}												
			}			
		}
		
		/* resumes a worker that is blocked */
		public void signal() {
			
			synchronized( this ) {
				
				_signalCount++;
				notifyAll();
			}
		}
		
		void execute(DispatcherQueue queue, Task task) {

			try {
				// execute the task and then enqueue any created subtasks
				Iterable<Task> children = task.execute();								
				 
				if( children != null ) {
					// enqueue all child tasks to the originating queue
					for( Task child : children ) {
						queue.enqueue(child);
					}										
				}
				
				/* notify any waiting threads, there is no parent-child
				 * relationship maintained between and their notifications.				 * 
				 */				
				synchronized( task ) { task.notifyAll(); }											
			}
			catch( Exception ex ) {

				Log.e("ThreadPool", ex);
			}			
		}
		
		DispatcherQueue nextQueue() {
			
			/* get the next queue and then dequeue the next task; both getNextQueue 
			 * and dequeue can potentially return null and therefore we use continue 
			 * to continue the while loop if null is returned
			 */
			DispatcherQueue queue = getNextQueue();
			assert( queue != null ); // this is probably a developer error... 
			return queue;
		}			
		
		/* blocks the current thread until signal is called on this object */
		void waitForSignal() {			
			
			try 
			{
				_idle = true;
				synchronized ( this )
				{// if there was a pending signal, decrement the signal counter
				 // and keep the thread running 
	
					if ( _signalCount == 0 )					
						wait();
					else
						_signalCount--;
				}
				
			} catch( InterruptedException iex ) {
				/* thread interrupted, resume as signaled, this will cause the 
				 * thread to attempt a poll of any dispatch queue. If no task
				 * is available for dispatching then it will return to blocked state;
				 * therefore this exception is "ignored"
				 */
				Log.i("ThreadPool.Worker", iex);			
			}			
			
			_spinCounter = 0;
			_idle = false;
		}
				
		boolean _idle = false;
		int _signalCount = 0;
		int _spinCounter = 0;
	}
	
	String _name;
	Worker[] _workers;
	AtomicBoolean _running = new AtomicBoolean(true);
	ArrayList<DispatcherQueue> _queues = new ArrayList<DispatcherQueue>();
	ReentrantLock _lock = new ReentrantLock(true);
	int _queueIndex = 0;
	
	static final int MAX_SPIN_COUNT = 1;
}
